package com.hubiwei.gmall.realtime.app.dws;

import com.hubiwei.gmall.realtime.app.BaseSqlApp;
import com.hubiwei.gmall.realtime.bean.KeyWordStats;
import com.hubiwei.gmall.realtime.common.Constant;
import com.hubiwei.gmall.realtime.function.IkAnalyzer;
import com.hubiwei.gmall.realtime.function.KwProduct;
import com.hubiwei.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/11/3 14:30
 */
public class DwsProductKeyWordApp extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsProductKeyWordApp().init(4005, 1, "DwsProductKeyWordApp");
    }
    
    @Override
    protected void run(StreamTableEnvironment tEnv) {
        // 1. 建动态表与 dwd_page进行关联
        tEnv.executeSql("create table product_stats(" +
                            "   stt string, " +
                            "   edt string," +
                            "   sku_name string," +
                            "   click_ct bigint," +
                            "   cart_ct bigint," +
                            "   order_ct bigint " +
                            ")with(" +
                            "   'connector'='kafka', " +
                            "   'properties.bootstrap.servers'='hadoop162:9092,hadoop163:9092,hadoop164:9092', " +
                            "   'properties.group.id'='DwsProductKeyWordApp', " +
                            "   'topic'='" + Constant.TOPIC_DWS_PRODUCT_STATS + "', " +
                            "   'scan.startup.mode'='latest-offset',  " +
                            "   'format'='json'  " +
                            ")");
        
        // 1. 过滤出来三个count至少一个不为0的数据
        Table t1 = tEnv.sqlQuery("select " +
                                     " * " +
                                     " from product_stats " +
                                     " where click_ct > 0 " +
                                     " or cart_ct > 0 " +
                                     " or order_ct > 0");
        tEnv.createTemporaryView("t1", t1);
        
        // 2. 把sku_name进行分词
        tEnv.createTemporaryFunction("ik_analyzer", IkAnalyzer.class);
        
        Table t2 = tEnv.sqlQuery("select" +
                                     " stt," +
                                     " edt, " +
                                     " word, " +
                                     " click_ct, " +
                                     " cart_ct, " +
                                     " order_ct " +
                                     "from t1 " +
                                     ", lateral table(ik_analyzer(sku_name))");
        tEnv.createTemporaryView("t2", t2);
        // 3. 按照窗口和关键词聚合
        Table t3 = tEnv.sqlQuery("select" +
                                     " stt, edt, word," +
                                     " sum(click_ct) click_ct, " +
                                     " sum(cart_ct) cart_ct, " +
                                     " sum(order_ct) order_ct " +
                                     "from t2 " +
                                     "group by stt, edt, word");
        tEnv.createTemporaryView("t3", t3);
        // 4. 把每行中的3个count, 变成三行
        tEnv.createTemporaryFunction("kw_product", KwProduct.class);
        
        Table result = tEnv.sqlQuery("select " +
                                         " stt, edt, word keyword," +
                                         " source, " +
                                         " ct, " +
                                         " unix_timestamp()*1000 ts " +
                                         " from t3 " +
                                         " join lateral table(kw_product(click_ct, cart_ct, order_ct)) on true");
        // 5. 写出到ClickHouse中
        tEnv
            .toRetractStream(result, KeyWordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink("gmall2021", "keyword_stats_2021", KeyWordStats.class));
        
    }
}
